#from pyspark import SparkContext,SparkConf
#from pyspark.sql import SparkSession
#conf = SparkConf( ).setAppName( "MySparkApp")
#sc=SparkContext( conf=conf)
#spark = SparkSession.builder.appName("PySparkExample").getOrCreate()

import pyspark.pandas as ps

# 忽略警告提示
import warnings
warnings.filterwarnings('ignore')
 
#导入处理数据包
import numpy as np
import pandas as pd
import time
import psutil

train = pd.read_csv("/home/ml/train.csv")
test  = pd.read_csv("/home/ml/test.csv")
#导入数据
#训练数据集
#train = spark.read.csv("file:///home/ml/train.csv")
#测试数据集
#test  = spark.read.csv("file:///home/ml/test.csv")
#print ('训练数据集:',train.shape,'测试数据集:',test.shape)

starttime = time.time()
#合并数据集，方便同时对两个数据集进行清洗
full = train._append( test , ignore_index = True )

print ('合并后的数据集:',full.shape)

#年龄(Age)
full['Age']=full['Age'].fillna( full['Age'].mean() )
#船票价格(Fare)
full['Fare'] = full['Fare'].fillna( full['Fare'].mean() )

'''
分类变量Embarked，看下最常见的类别，用其填充
'''
full['Embarked'].value_counts()
'''
从结果来看，S类别最常见。我们将缺失值填充为最频繁出现的值：
S=英国南安普顿Southampton
'''
full['Embarked'] = full['Embarked'].fillna( 'S' )

#船舱号（Cabin）：查看里面数据长啥样
full['Cabin'].head()

#缺失数据比较多，船舱号（Cabin）缺失值填充为U，表示未知（Uknow）
full['Cabin'] = full['Cabin'].fillna( 'U' )

'''
将性别的值映射为数值
男（male）对应数值1，女（female）对应数值0
'''
sex_mapDict={'male':1,
            'female':0}
#map函数：对Series每个数据应用自定义的函数计算
full['Sex']=full['Sex'].map(sex_mapDict)

#存放提取后的特征
embarkedDf = pd.DataFrame()

'''
使用get_dummies进行one-hot编码，产生虚拟变量（dummy variables），列名前缀是Embarked
'''
embarkedDf = pd.get_dummies( full['Embarked'] , prefix='Embarked' )

#添加one-hot编码产生的虚拟变量（dummy variables）到泰坦尼克号数据集full
full = pd.concat([full,embarkedDf],axis=1)

'''
因为已经使用登船港口(Embarked)进行了one-hot编码产生了它的虚拟变量（dummy variables）
所以这里把登船港口(Embarked)删掉
'''
full.drop('Embarked',axis=1,inplace=True)
'''
上面drop删除某一列代码解释：
因为drop(name,axis=1)里面指定了name是哪一列，比如指定的是A这一列，axis=1表示按行操作。
那么结合起来就是把A列里面每一行删除，最终结果是删除了A这一列.
简单来说，使用drop删除某几列的方法记住这个语法就可以了：drop([列名1,列名2],axis=1)
'''

'''
客舱等级(Pclass):
1=1等舱，2=2等舱，3=3等舱
'''
#存放提取后的特征
pclassDf = pd.DataFrame()

#使用get_dummies进行one-hot编码，列名前缀是Pclass
pclassDf = pd.get_dummies( full['Pclass'] , prefix='Pclass' )
pclassDf.head()

#添加one-hot编码产生的虚拟变量（dummy variables）到泰坦尼克号数据集full
full = pd.concat([full,pclassDf],axis=1)

#删掉客舱等级（Pclass）这一列
full.drop('Pclass',axis=1,inplace=True)

'''
定义函数：从姓名中获取头衔
'''
def getTitle(name):
    str1=name.split( ',' )[1] #Mr. Owen Harris
    str2=str1.split( '.' )[0]#Mr
    #strip() 方法用于移除字符串头尾指定的字符（默认为空格）
    str3=str2.strip()
    return str3

#存放提取后的特征
titleDf = pd.DataFrame()
#map函数：对Series每个数据应用自定义的函数计算
titleDf['Title'] = full['Name'].map(getTitle)

'''
定义以下几种头衔类别：
Officer政府官员
Royalty王室（皇室）
Mr已婚男士
Mrs已婚妇女
Miss年轻未婚女子
Master有技能的人/教师
'''
#姓名中头衔字符串与定义头衔类别的映射关系
title_mapDict = {
                    "Capt":       "Officer",
                    "Col":        "Officer",
                    "Major":      "Officer",
                    "Jonkheer":   "Royalty",
                    "Don":        "Royalty",
                    "Sir" :       "Royalty",
                    "Dr":         "Officer",
                    "Rev":        "Officer",
                    "the Countess":"Royalty",
                    "Dona":       "Royalty",
                    "Mme":        "Mrs",
                    "Mlle":       "Miss",
                    "Ms":         "Mrs",
                    "Mr" :        "Mr",
                    "Mrs" :       "Mrs",
                    "Miss" :      "Miss",
                    "Master" :    "Master",
                    "Lady" :      "Royalty"
                    }

#map函数：对Series每个数据应用自定义的函数计算
titleDf['Title'] = titleDf['Title'].map(title_mapDict)

#使用get_dummies进行one-hot编码
titleDf = pd.get_dummies(titleDf['Title'])
titleDf.head()

#添加one-hot编码产生的虚拟变量（dummy variables）到泰坦尼克号数据集full
full = pd.concat([full,titleDf],axis=1)

#删掉姓名这一列
full.drop('Name',axis=1,inplace=True)

#存放客舱号信息
cabinDf = pd.DataFrame()

# '''
# 客场号的类别值是首字母，例如：
# C85 类别映射为首字母C
# '''
full[ 'Cabin' ] = full[ 'Cabin' ].map( lambda c : c[0] )

##使用get_dummies进行one-hot编码，列名前缀是Cabin
cabinDf = pd.get_dummies( full['Cabin'] , prefix = 'Cabin' )

#添加one-hot编码产生的虚拟变量（dummy variables）到泰坦尼克号数据集full
full = pd.concat([full,cabinDf],axis=1)

#删掉客舱号这一列
full.drop('Cabin',axis=1,inplace=True)

#删掉船票这一列
full.drop('Ticket',axis=1,inplace=True)

#存放家庭信息
familyDf = pd.DataFrame()

# '''
# 家庭人数=同代直系亲属数（Parch）+不同代直系亲属数（SibSp）+乘客自己
# （因为乘客自己也是家庭成员的一个，所以这里加1）
# '''
familyDf[ 'FamilySize' ] = full[ 'Parch' ] + full[ 'SibSp' ] + 1

# '''
# 家庭类别：
# 小家庭Family_Single：家庭人数=1
# 中等家庭Family_Small: 2<=家庭人数<=4
# 大家庭Family_Large: 家庭人数>=5
# '''
#if 条件为真的时候返回if前面内容，否则返回0
familyDf[ 'Family_Single' ] = familyDf[ 'FamilySize' ].map( lambda s : 1 if s == 1 else 0 )
familyDf[ 'Family_Small' ]  = familyDf[ 'FamilySize' ].map( lambda s : 1 if 2 <= s <= 4 else 0 )
familyDf[ 'Family_Large' ]  = familyDf[ 'FamilySize' ].map( lambda s : 1 if 5 <= s else 0 )

familyDf.head()

#添加one-hot编码产生的虚拟变量（dummy variables）到泰坦尼克号数据集full
full = pd.concat([full,familyDf],axis=1)

#相关性矩阵
corrDf = full.corr()
# '''
# 查看各个特征与生成情况（Survived）的相关系数，
# ascending=False表示按降序排列
# '''
corrDf['Survived'].sort_values(ascending =False)

#特征选择
full_X = pd.concat( [titleDf,#头衔
                     pclassDf,#客舱等级
                     familyDf,#家庭大小
                     full['Fare'],#船票价格
                     cabinDf,#船舱号
                     embarkedDf,#登船港口
                     full['Sex']#性别
                    ] , axis=1 )

#原始数据集有891行
sourceRow=891

# '''
# sourceRow是我们在最开始合并数据前知道的，原始数据集有总共有891条数据
# 从特征集合full_X中提取原始数据集提取前891行数据时，我们要减去1，因为行号是从0开始的。
# '''
#原始数据集：特征
source_X = full_X.loc[0:sourceRow-1,:]
#原始数据集：标签
source_y = full.loc[0:sourceRow-1,'Survived']

#预测数据集：特征
pred_X = full_X.loc[sourceRow:,:]
# '''
# 上面代码解释：
# 891行前面的数据是测试数据集，891行之后的数据是预测数据集。[sourceRow:,:]就是从891行开始到最后一行作为预测数据集
# '''

# '''
# 确保这里原始数据集取的是前891行的数据，不然后面模型会有错误
# '''
#原始数据集有多少行
print('原始数据集有多少行:',source_X.shape[0])
#预测数据集大小
print('原始数据集有多少行:',pred_X.shape[0])

# '''
# 从原始数据集（source）中拆分出训练数据集（用于模型训练train），测试数据集（用于模型评估test）
# train_test_split是交叉验证中常用的函数，功能是从样本中随机的按比例选取train data和test data
# train_data：所要划分的样本特征集
# train_target：所要划分的样本结果
# test_size：样本占比，如果是整数的话就是样本的数量
# '''
from sklearn.model_selection import train_test_split

#建立模型用的训练数据集和测试数据集
train_X, test_X, train_y, test_y = train_test_split(source_X ,
                                                    source_y,
                                                    train_size=.8)

#输出数据集大小
print ('原始数据集特征：',source_X.shape,
       '训练数据集特征：',train_X.shape ,
      '测试数据集特征：',test_X.shape)

print ('原始数据集标签：',source_y.shape,
       '训练数据集标签：',train_y.shape ,
      '测试数据集标签：',test_y.shape)

# 随机森林Random Forests Model
from sklearn.ensemble import RandomForestClassifier
model = RandomForestClassifier(n_estimators=100)

#第3步：训练模型
model.fit( train_X , train_y )

# 获取CPU使用率
cpu_usage = psutil.cpu_percent(interval=1)

# 获取RAM使用情况
ram_usage = psutil.virtual_memory()

print("当前CPU使用率：%f%%" % cpu_usage)
print("总内存：%dMB" % (ram_usage.total / 1024 / 1024))
print("可用内存：%dMB" % (ram_usage.available / 1024 / 1024))
print("已用内存：%dMB" % (ram_usage.used / 1024 / 1024))
print("内存使用率：%d%%" % ram_usage.percent)

# 分类问题，score得到的是模型的正确率
model.score(test_X , test_y )

# 随机森林的模型报告
from sklearn.metrics import classification_report
from sklearn.metrics import confusion_matrix
from sklearn.metrics import roc_curve, auc

y_predict = model.predict(test_X)
print('召回率等评分')
print(classification_report(y_predict, test_y, target_names=['died','survived']))
# print('混淆矩阵')
# print(confusion_matrix(y_predict, test_y))
endtime = time.time()
print('运行时间',endtime-starttime)
